package com.study.chapter09;

import com.study.entity.Action;
import com.study.entity.Pattern;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/17 23:03
 */
public class CheckpointSetTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //开启检查点
        env.enableCheckpointing(1000L);
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        // 配置存储检查点到JobManager堆内存
        env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
        // 配置存储检查点到文件系统
        env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://namenode:40010/flink/checkpoints"));

        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 设置精确一次模式
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 最小间隔时间500毫秒
        checkpointConfig.setMinPauseBetweenCheckpoints(500);
        // 超时时间1分钟
        checkpointConfig.setCheckpointTimeout(60000);
        // 同时只能有一个检查点
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        // 开启检查点的外部持久化保存，作业取消后依然保留
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 启用不对齐的检查点保存方式
        checkpointConfig.enableUnalignedCheckpoints();
        // 设置检查点存储，可以直接传入一个String，指定文件系统的路径
        checkpointConfig.setCheckpointStorage("hdfs://my/checkpoint/dir");


        DataStreamSource<Action> actionStream = env.fromElements(
                new Action("Bob", "login"),
                new Action("Bob", "pay"),
                new Action("Alice", "login"),
                new Action("Alice", "order")
        );


        env.execute();
    }

}
